package thread

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"runtime/debug"
	"sync"
	"time"

	"gitee.com/baixudong/gospider/chanx"
)

type Client struct {
	Debug     bool               //是否显示调试信息
	ctx2      context.Context    //控制各个协程
	cnl2      context.CancelFunc //控制各个协程
	ctx       context.Context    //控制主进程，不会关闭各个协程
	cnl       context.CancelFunc //控制主进程，不会关闭各个协程
	ctx3      context.Context    //chanx 的协程控制
	cnl3      context.CancelFunc //chanx 的协程控制
	tasks     chan *Task
	tasks2    *chanx.Client[*Task] //chanx 的队列任务
	sones     chan struct{}
	runNum    int //正在运行的任务数量
	threadNum int //正在运行的协程数量
	taskNum   int //剩余任务数量
	maxNum    int //最大任务数量
	timeOut   int
	lock      sync.Mutex
	callBack  func(*Task) error //任务回调
	err       error
}
type Task struct {
	Func     any                                //运行的函数
	Args     []any                              //传入的参数
	CallBack func(context.Context, []any) error //回调函数
	Timeout  int                                //超时时间
	Result   []any                              //函数执行的结果
	Error    error                              //函数错误信息
	Done     chan struct{}                      //等待完成
}
type ClientOption struct {
	Timeout  int               //任务超时时间
	CallBack func(*Task) error //任务回调
}

func NewClient(preCtx context.Context, maxNum int, options ...ClientOption) *Client {
	if preCtx == nil {
		preCtx = context.TODO()
	}
	if maxNum < 1 {
		panic("协程池不能为空")
	}
	var option ClientOption
	if len(options) > 0 {
		option = options[0]
	}
	if option.Timeout <= 0 {
		option.Timeout = 60
	}
	ctx2, cnl2 := context.WithCancel(preCtx)
	ctx, cnl := context.WithCancel(preCtx)
	tasks := make(chan *Task)
	sones := make(chan struct{})
	pool := &Client{
		callBack: option.CallBack, timeOut: option.Timeout,
		ctx2: ctx2, cnl2: cnl2, ctx: ctx, cnl: cnl,
		tasks: tasks, sones: sones, maxNum: maxNum,
	}
	if option.CallBack != nil {
		ctx3, cnl3 := context.WithCancel(preCtx)
		pool.tasks2 = chanx.NewClient[*Task](preCtx)
		pool.ctx3 = ctx3
		pool.cnl3 = cnl3
		go pool.taskMain2()
	}
	go pool.taskMain()
	return pool
}
func (obj *Client) TaskSize() int { //剩余任务数量，包括正在运行的任务
	return obj.taskNum
}
func (obj *Client) caseMain() {
	obj.lock.Lock()
	obj.taskNum++
	if obj.threadNum < obj.maxNum && obj.runNum < obj.taskNum {
		obj.threadNum++
		go obj.runMain()
	}
	obj.lock.Unlock()
}

func (obj *Client) taskMain() {
	defer close(obj.sones)
	defer close(obj.tasks)
	defer obj.cnl2()
	for {
		select {
		case <-obj.ctx2.Done():
			return
		case <-obj.ctx.Done():
			for {
				select {
				case <-obj.ctx2.Done():
					return
				case <-obj.sones:
					obj.caseMain()
				default:
					if obj.taskNum <= 0 {
						return
					}
					time.Sleep(time.Duration(1) * time.Second)
				}
			}
		case <-obj.sones:
			obj.caseMain()
		}
	}
}

func (obj *Client) taskMain2() {
	defer obj.cnl3()
	defer obj.Close()
	defer obj.tasks2.Close()
	for task := range obj.tasks2.Chan() {
		<-task.Done
		if task.Error != nil {
			obj.err = task.Error
			return
		}
		if err := obj.callBack(task); err != nil {
			obj.err = err
			return
		}
	}
}
func (obj *Client) subThreadNum() {
	obj.lock.Lock()
	obj.threadNum--
	obj.lock.Unlock()
}
func (obj *Client) caseTask(task *Task) {
	obj.lock.Lock()
	obj.runNum++
	obj.lock.Unlock()
	obj.run(task)
	obj.lock.Lock()
	obj.runNum--
	obj.taskNum--
	obj.lock.Unlock()
}
func (obj *Client) runMain() {
	defer obj.subThreadNum()
	for {
		select {
		case <-obj.ctx2.Done():
			return
		case <-obj.ctx.Done():
			for {
				select {
				case <-obj.ctx2.Done():
					return
				case task := <-obj.tasks:
					if task == nil {
						return
					}
					obj.caseTask(task)
				default:
					if obj.taskNum <= obj.runNum {
						return
					}
				}
			}
		case task := <-obj.tasks:
			if task == nil {
				return
			}
			obj.caseTask(task)
		case <-time.After(time.Second * time.Duration(obj.timeOut)):
			return
		}
	}
}

var ErrPoolClosed = errors.New("pool closed")

func (obj *Client) verify(fun any, args []any) error {
	if fun == nil {
		return errors.New("not func")
	}
	typeOfFun := reflect.TypeOf(fun)
	if typeOfFun.Kind() != reflect.Func {
		return errors.New("not func")
	}
	if typeOfFun.NumIn() != len(args)+1 {
		return errors.New("args num error")
	}
	if typeOfFun.In(0).String() != "context.Context" {
		return errors.New("frist params not context.Context")
	}
	for i := 1; i < len(args)+1; i++ {
		if args[i-1] == nil {
			if reflect.Zero(typeOfFun.In(i)).Interface() != args[i-1] {
				return errors.New("args type not equel")
			}
		} else if !reflect.TypeOf(args[i-1]).ConvertibleTo(typeOfFun.In(i)) {
			return errors.New("args type not equel")
		}
	}
	return nil
}

// 创建task
func (obj *Client) Write(task *Task) (*Task, error) {
	task.Done = make(chan struct{})
	if err := obj.verify(task.Func, task.Args); err != nil {
		task.Error = err
		close(task.Done)
		return task, err
	}
	select {
	case <-obj.ctx2.Done():
		if obj.err != nil {
			task.Error = obj.err
		} else {
			task.Error = ErrPoolClosed
		}
		close(task.Done)
		return task, task.Error
	case <-obj.ctx.Done():
		if obj.err != nil {
			task.Error = obj.err
		} else {
			task.Error = ErrPoolClosed
		}
		close(task.Done)
		return task, task.Error
	case obj.sones <- struct{}{}:
		select {
		case <-obj.ctx2.Done():
			if obj.err != nil {
				task.Error = obj.err
			} else {
				task.Error = ErrPoolClosed
			}
			close(task.Done)
			return task, task.Error
		case <-obj.ctx.Done():
			if obj.err != nil {
				task.Error = obj.err
			} else {
				task.Error = ErrPoolClosed
			}
			close(task.Done)
			return task, task.Error
		case obj.tasks <- task:
			if obj.tasks2 != nil {
				obj.tasks2.Add(task)
			}
			return task, obj.err
		}
	}
}
func (obj *Client) run(task *Task) {
	defer close(task.Done)
	defer func() {
		if r := recover(); r != nil {
			task.Error = fmt.Errorf("%v", r)
			if obj.Debug {
				debug.PrintStack()
			}
		}
	}()
	timeOut := task.Timeout
	var ctx context.Context
	var cnl context.CancelFunc
	if timeOut > 0 {
		ctx, cnl = context.WithTimeout(obj.ctx2, time.Second*time.Duration(timeOut))
	} else {
		ctx, cnl = context.WithCancel(obj.ctx2)
	}
	defer cnl()
	params := make([]reflect.Value, len(task.Args)+1)
	params[0] = reflect.ValueOf(ctx)
	for k, param := range task.Args {
		params[k+1] = reflect.ValueOf(param)
	}
	task.Result = []any{}
	for _, rs := range reflect.ValueOf(task.Func).Call(params) { //执行主方法
		task.Result = append(task.Result, rs.Interface())
	}
	if task.CallBack != nil {
		task.Error = task.CallBack(ctx, task.Result) //执行回调方法
	}
}

func (obj *Client) Join() error { //等待所有任务完成，并关闭pool
	obj.cnl()
	if obj.tasks2 != nil {
		obj.tasks2.Join()
		<-obj.ctx3.Done()
	}
	<-obj.ctx2.Done() //等待所有的任务执行完毕
	return obj.err
}

func (obj *Client) Close() { //告诉所有协程，立即结束任务
	obj.cnl()
	obj.cnl2()
	if obj.tasks2 != nil {
		obj.tasks2.Close()
	}
}
func (obj *Client) Err() error { //错误
	return obj.err
}
func (obj *Client) Done() <-chan struct{} { //所有任务执行完毕
	return obj.ctx2.Done()
}
func (obj *Client) RunSize() int { //运行的协程数量
	return obj.runNum
}
func (obj *Client) ThreadSize() int { //创建的协程数量
	return obj.threadNum
}
func (obj *Client) Empty() bool { //任务是否为空
	if obj.RunSize() <= 0 && obj.TaskSize() <= 0 && obj.ThreadSize() <= 0 {
		return true
	}
	return false
}
